package main

import (
	"fmt"
	"github.com/Shopify/sarama"
	"strings"
	"sync"
	"time"
)

type XLConsumer struct {
	offsetCatch *cluster.OffsetStash
	consumerObj *cluster.Consumer
	dataCB      FuncDataCallback
	eventCB     FuncEventCallback
	bExit       bool
	wg          *sync.WaitGroup
	bPause      bool
}

func (this *XLConsumer) Init(conf *KafkaCfg, dataCB FuncDataCallback) (err error) {
	this.dataCB = dataCB
	this.eventCB = nil
	this.offsetCatch = cluster.NewOffsetStash()
	config := cluster.NewConfig()
	config.Group.Return.Notifications = true
	config.Consumer.Offsets.CommitInterval = 1 * time.Second
	config.Version = sarama.V0_10_2_1

	// 初始从最新的offset开始,正式上线后改为OffsetNewest
	//config.Consumer.Offsets.Initial = sarama.OffsetNewest
	switch conf.Consumer.InitOffset {
	case "Newest":
		config.Consumer.Offsets.Initial = sarama.OffsetNewest
	case "Oldest":
		config.Consumer.Offsets.Initial = sarama.OffsetOldest
	default:
		config.Consumer.Offsets.Initial = sarama.OffsetNewest
	}

	fmt.Println("---------- 主题：", conf.Consumer.Topics, "kafkaBroker:", conf.Consumer.Brokers)

	//	连接kafka
	this.consumerObj, err = cluster.NewConsumer(
		strings.Split(conf.Consumer.Brokers, ","),
		conf.Consumer.GroupID,
		strings.Split(conf.Consumer.Topics, ","),
		config)

	if err != nil {
		return err
	}

	this.bExit = false
	this.bPause = false

	this.wg = &sync.WaitGroup{}
	this.wg.Add(3)
	go this.dealDataThread()
	go this.dealEventThread()
	go this.commitOffsetThread()

	return nil
}

func (this *XLConsumer) InitWithSASL(conf *KafkaCfg, sasl *SASLCfg, dataCB FuncDataCallback, eventCB FuncEventCallback) (err error) {

	this.dataCB = dataCB
	this.eventCB = eventCB
	this.offsetCatch = cluster.NewOffsetStash()

	config := cluster.NewConfig()
	config.Group.Return.Notifications = true
	config.Consumer.Offsets.CommitInterval = 1 * time.Second
	switch conf.Consumer.InitOffset {
	case "Newest":
		config.Consumer.Offsets.Initial = sarama.OffsetNewest
	case "Oldest":
		config.Consumer.Offsets.Initial = sarama.OffsetOldest
	default:
		config.Consumer.Offsets.Initial = sarama.OffsetNewest
	}

	// sasl相关配置
	if sasl != nil {
		config.Net.SASL.Enable = true
		config.Net.SASL.Handshake = true
		config.Net.SASL.User = sasl.UserName
		config.Net.SASL.Password = sasl.PassWord
	}

	fmt.Println("---------- 主题：", conf.Consumer.Topics, "kafkaBroker:", conf.Consumer.Brokers)

	//	连接kafka
	this.consumerObj, err = cluster.NewConsumer(
		strings.Split(conf.Consumer.Brokers, ","),
		conf.Consumer.GroupID,
		strings.Split(conf.Consumer.Topics, ","),
		config)

	if err != nil {
		return err
	}

	this.bExit = false
	this.bPause = false

	this.wg = &sync.WaitGroup{}
	this.wg.Add(3)
	go this.dealDataThread()
	go this.dealEventThread()
	go this.commitOffsetThread()

	return nil
}

func (this *XLConsumer) Fini() error {
	this.bExit = true
	this.wg.Wait()
	return this.consumerObj.Close()
}

func (this *XLConsumer) SetEventCB(pFunc FuncEventCallback) {
	this.eventCB = pFunc
}

func (this *XLConsumer) MarkOffset(msg *sarama.ConsumerMessage) {
	this.offsetCatch.MarkOffset(msg, "")
}

func (this *XLConsumer) dealDataThread() {
	defer this.wg.Done()
	for {
		if this.bExit == true {
			fmt.Println("[dealDataThread] 数据处理线程退出！")
			break
		}

		select {
		case msg := <-this.consumerObj.Messages():
			if this.dataCB != nil {
				this.dataCB(msg)
			}
		case <-time.After(1 * time.Second):
			time.Sleep(time.Millisecond * 1)
		}
	}
}

func (this *XLConsumer) dealEventThread() {
	defer this.wg.Done()
	for {
		if this.bExit == true {
			fmt.Println("[dealDataThread] 事件处理线程退出！")
			break
		}

		if this.bPause == true {
			time.Sleep(time.Second)
			continue
		}

		select {
		case err := <-this.consumerObj.Errors():
			if err != nil {
				errStr := fmt.Sprintf("kafka消费者异常，错误：%v", err.Error())
				if this.eventCB != nil {
					this.eventCB(errStr, "Error")
				} else {
					fmt.Println(errStr)
				}
			}
		case msg := <-this.consumerObj.Notifications():
			infoStr := fmt.Sprintf("\nRebalance Type:%v\nClaimed:%v\nReleased:%v\nCurrent:%v", msg.Type, msg.Claimed, msg.Released, msg.Current)
			if this.eventCB != nil {
				this.eventCB(infoStr, "Info")
			} else {
				fmt.Println(infoStr)
			}
		case <-time.After(1 * time.Second):
			time.Sleep(time.Millisecond * 1)
		}
		time.Sleep(time.Millisecond * 10)
	}
}

func (this *XLConsumer) commitOffsetThread() {
	defer this.wg.Done()
	commitTime := time.Now()
	for {
		if this.bExit == true {
			//退出前提交Offset
			err := this.consumerObj.CommitOffsets()
			if err != nil {
				fmt.Println("[commitOffsetThread] 提交offset错误!err=", err)
			}
			fmt.Println("[commitOffsetThread] offset提交线程退出！")
			break
		}

		if time.Since(commitTime) > time.Second {
			if this.offsetCatch != nil {
				offsets := this.offsetCatch.Offsets()
				if len(offsets) != 0 {
					fmt.Println("[commitOffsetThread] 提交offset!", offsets)
					this.consumerObj.MarkOffsets(this.offsetCatch)
					err := this.consumerObj.CommitOffsets()
					if err != nil {
						fmt.Println("[commitOffsetThread] 提交offset错误!err=", err)
					}
				}
			}
			commitTime = time.Now()
		}

		time.Sleep(time.Millisecond * 100)
	}
}

func (this *XLConsumer) CommitOffset() error {
	offsets := this.offsetCatch.Offsets()
	if len(offsets) != 0 {
		fmt.Println("[commitOffsetThread] 提交offset!", offsets)
		this.consumerObj.MarkOffsets(this.offsetCatch)
		err := this.consumerObj.CommitOffsets()
		if err != nil {
			fmt.Println("[commitOffsetThread] 提交offset错误!err=", err)
			return err
		}
	}

	return nil
}

func (this *XLConsumer) Pause() {
	this.bPause = true
}

func (this *XLConsumer) Resume() {
	this.bPause = false

}
